package com.hp.consumer;

import com.hp.consumer.config.User;
import com.hp.consumer.config.UserRepository;
import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;
import java.util.function.Consumer;

@SpringBootApplication
//@EnableBinding({PropertiesInSource.class})
@RequiredArgsConstructor(onConstructor_ = {@Autowired})
@RestController
@Slf4j
public class SpringKafkaConsumerApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringKafkaConsumerApplication.class, args);
    }

//    @StreamListener(PropertiesInSource. DEMO_INPUT)
//    public void insertQuotationK(Message<String> message) {
//        if (StringUtils.isEmpty(mess  age.getPayload())) {
//            System.out.println("receiver data is empty !");
//            System.out.println(400 + "failed");
//        }
//        System.out.println("kafka收到"+message.getPayload());
//    }

    private final UserRepository repository;
    private final MongoTemplate mongoTemplate;

//    @StreamListener(PropertiesInSource. DEMO_INPUT)
//    public void insertQuotationK(Message<User> message) {
//        User user = message.getPayload();
//        System.out.println("kafka收到"+ JSON.toJSONString(user));
//        System.out.println("kafka收到"+ user);
////        repository.save(user);
//    }


    //@StreamListener(PropertiesInSource. DEMO_INPUT)
    public void insertQuotationK(Message<String> message) {
        System.out.println("kafka收到" + message.getPayload());
    }

//    @PostMapping("/user")
//    public User createUser(@RequestBody User user) {
//        user.setId("111111122221");
////        User save = repository.save(user);
//        mongoTemplate.insert(user);
//        return user;
//
//    }


    @Bean
    public Consumer<Message<User>> inputKafka1() {
        return str -> {
            // 收到消息在这里做一些处理
            log.info("inputKafka1 message: {}", str);
            // 配置文件 必须开启对应的 配置信息,才能 在 header 上 获取到数据
            Acknowledgment acknowledgment = str.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
            if (acknowledgment != null) {
                System.out.println("Acknowledgment provided");
                acknowledgment.acknowledge();
            }

        };
    }

    @Bean
    public Consumer<User> inputKafka2() {
        return str -> {
            log.info("inputKafka2 message: {}", str);
        };
    }

    /**
     * ① basicReject(deliveryTag, true): 拒绝deliveryTag对应的消息,true则重新入队列，否则丢弃或者进入死信队列. 该方法reject后，该消费者还是会消费到该条被reject的消息
     * ② basicNack(deliveryTag, false, true): 不确认deliveryTag对应的消息，第二个参数是否应用于多消息，第三个参数是否requeue. 与basic.reject区别就是同时支持多个消息，可以nack该消费者先前接收未ack的所有消息。nack后的消息也会被自己消费到
     * ③ basicRecover(true): basic.recover是否恢复消息到队列，参数是是否requeue，true则重新入队列，并且尽可能的将之前recover的消息投递给其他消费者消费，而不是自己再次消费。false则消息会重新被投递给自己。
     *
     * @return {@link Consumer}<{@link Message}<{@link User}>>
     */
    @Bean
    public Consumer<Message<User>> inputRabbit1() {
        return str -> {
            log.info("inputRabbit1 message: {}", str);
            // 配置文件 必须开启对应的 配置信息,才能 在 header 上 获取到数据
            Channel channel = str.getHeaders().get(AmqpHeaders.CHANNEL, Channel.class);
            Long deliveryTag = str.getHeaders().get(AmqpHeaders.DELIVERY_TAG, Long.class);
            try {
                final User user = str.getPayload();
                if (user.getAge() < 5) {
                    // 消息确认,deliveryTag:该消息的index,multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息
                    channel.basicAck(deliveryTag, false);
                } else {
                    // 重回队列,拒绝一条消息
                    // deliveryTag:该消息的index
                    // requeue：被拒绝的是否重新入队列
                    // 该方法reject后，该消费者还是会消费到该条被reject的消息
                    channel.basicReject(deliveryTag, true);
                    // true则重新入队列，并且尽可能的将之前recover的消息投递给其他消费者消费，而不是自己再次消费。false则消息会重新被投递给自己。
                    channel.basicRecover(true);
                }
            } catch (Exception e) {
                try {
                    //消费者处理出了问题，需要告诉队列信息消费失败
                    // deliveryTag:该消息的index
                    // multiple：是否批量.true:将一次性拒绝所有小于deliveryTag的消息
                    // requeue：被拒绝的是否重新入队列
                    channel.basicNack(deliveryTag, false, true);
                } catch (IOException ex) {
                    throw new RuntimeException(ex);
                }

            }
        };
    }

    ///**
    // * S1,S2,S3。S1丢给S2一个消息，S2也不消费，只是要把它丢给S3，那么S2中就可以使用Function，将接到的消息转化成S3需要的格式，然后丢过去.
    // * <p>
    // * 总之一个原则，java.util.function.Function 要求必须有输入destination和输出destination，需要你在配置文件里配置.
    // * <p>
    // * 下面的函数就将自己接收到的消息，从MsgData转化为String然后丢给下一个Consumer去消费
    // *
    // * @return {@link Function}<{@link Message}<{@link User}>, {@link Message}<{@link String}>>
    // */
    //@Bean
    //public Function<Message<User>, Message<String>> ss007Function() {
    //    return new Function<Message<User>, Message<String>>() {
    //        @Override
    //        public Message<String> apply(Message<User> msg) {
    //            User payload = msg.getPayload();
    //            return MessageBuilder.withPayload(String.format("王二狗第%d说：%s", payload.getId(), payload))
    //                    .build();
    //        }
    //    };
    //}

    @Bean
    public Consumer<User> inputRabbit2() {
        return str -> {
            log.info("inputRabbit2 message: {}", str);
        };
    }

}
